package com.lubase.metadata.consumer;

import com.alibaba.fastjson.JSON;
import com.lubase.metadata.service.ApplyLakeService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.annotation.SelectorType;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@RocketMQMessageListener(topic = "LCP_WF_PUBLIC_TOPIC", selectorExpression = "1481919486398304256", selectorType = SelectorType.TAG, consumerGroup = "apply_lake_flow_end_group")
@Slf4j
@Component
public class ApplyLakeFlowEnd implements RocketMQListener<String> {

    @Autowired
    ApplyLakeService applyLakeService;

    @Override
    public void onMessage(String s) {
        if (StringUtils.isEmpty(s)) {
            return;
        }
        log.info("ApplyLakeFlowEnd onMessage:{}", s);
        try {
            // 反序列 WFEventModel对象
            WFEventModel wfEventModel = JSON.parseObject(s, WFEventModel.class);
            if (wfEventModel == null) {
                return;
            }
            if (wfEventModel.getOpenEventType().equals(EOpenEventType.End)) {
                // 流程结束时把状态更新为入湖成功
            }
            String lakeId = wfEventModel.getDataId();
            applyLakeService.applyLakeComplete(lakeId);
        } catch (Exception e) {
            log.error("ApplyLakeFlowEnd onMessage error:{}", e);
        }
    }

}
